package com.navinfo.platform.trip.analysis.flink.trigger;

import com.navinfo.platform.trip.analysis.flink.config.SystemConfig;
import com.navinfo.platform.trip.common.enums.SystemTypeEnum;
import com.navinfo.platform.trip.analysis.pojo.TripJudgeState;
import com.navinfo.platform.trip.analysis.service.ServiceFactory;
import com.navinfo.platform.trip.analysis.service.TripJudgeService;
import com.navinfo.platform.trip.common.pojo.PointProtocol;
import com.navinfo.platform.trip.common.enums.ProtocolEnum;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.java.typeutils.PojoTypeInfo;
import org.apache.flink.api.java.utils.ParameterTool;
import org.apache.flink.streaming.api.datastream.CoGroupedStreams;
import org.apache.flink.streaming.api.windowing.triggers.Trigger;
import org.apache.flink.streaming.api.windowing.triggers.TriggerResult;
import org.apache.flink.streaming.api.windowing.windows.GlobalWindow;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.Optional;

/**
 * 行程切分触发器
 * @author: web
 * @data: 2020-04-07
 **/
@SuppressWarnings("unchecked")
public class TripTrigger<IN> extends Trigger<IN, GlobalWindow> {

    private static final Logger logger = LoggerFactory.getLogger(TripTrigger.class);

    /**
     * 最小速度，单位km/h
     */
    private int speedMin;
    /**
     * 最大速度，单位km/h
     */
    private int speedMax;
    /**
     * 引擎最小转速
     */
    private int rotationMin;
    /**
     * 引擎最大转速
     */
    private int rotationMax;

    /*
     *行程时长小于多少秒不保存，单位s
     */
    private int tripMinDuration;

    /**
     * 行程切分服务
     */
    private TripJudgeService tripJudgeService;

    /**
     * 行程状态判断的缓存数据
     */
    private final ValueStateDescriptor<TripJudgeState> tripJudgeDesc = new ValueStateDescriptor<>("trip_judge_state", PojoTypeInfo.of(TripJudgeState.class));

    public TripTrigger(ParameterTool jobConfig) {
        rotationMin = jobConfig.getInt("rotation_min", 0);
        rotationMax = jobConfig.getInt("rotation_max", 10000);
        speedMin = jobConfig.getInt("speed_min", 0);
        speedMax = jobConfig.getInt("speed_max", 200);
        tripMinDuration = jobConfig.getInt(SystemConfig.tripMinDurationKey, 120);

        SystemTypeEnum systemType = SystemTypeEnum.valueOf(jobConfig.getInt(SystemConfig.systemTypeKey));
        tripJudgeService = ServiceFactory.singleServiceImpl(systemType, TripJudgeService.class);
    }

    /**
     * 双流中每到一个元素都会进入该方法，每次只有一个元素，element.isOne为True或element.isTwo为True
     * @param element   传入元素
     * @param timestamp 事件时间，Source指定的事件时间提取器
     * @param window    窗口对象
     * @param ctx       上下文
     * @return 是否触发
     */
    @Override
    public TriggerResult onElement(IN element, long timestamp, GlobalWindow window, TriggerContext ctx) throws Exception {
        if(element instanceof PointProtocol){
            return getTriggerResult((PointProtocol) element, ctx);
        }else if(element instanceof CoGroupedStreams.TaggedUnion){
            return getTriggerResult((CoGroupedStreams.TaggedUnion<PointProtocol, PointProtocol>) element, ctx);
        }else{
            logger.error("TripTrigger处理的元素类型不匹配，默认跳过...");
        }

        return TriggerResult.CONTINUE;
    }

    /**
     * 处理通过Union连接的0F37和0200两个流
     * @param element 公共轨迹点
     * @param ctx 上下文
     * @return 触发结果
     */
    private TriggerResult getTriggerResult(PointProtocol element, TriggerContext ctx) throws java.io.IOException {
        //使用0200判断行程，0F37点跳过
        if(ProtocolEnum.REALTIME.equals(element.getProtocol())){
            return TriggerResult.CONTINUE;
        }

        //数据合法性校验
        if(!checkData(element)){
            return TriggerResult.CONTINUE;
        }

        //缓存的行程状态判断数据
        ValueState<TripJudgeState> tripJudgeState = ctx.getPartitionedState(tripJudgeDesc);
        TripJudgeState tripJudge = tripJudgeState.value();
        long currGpsTime = element.getGpsTime();

        //行程状态处理
        TriggerResult result;
        switch (tripJudgeService.tripStatus(tripJudge, element)) {
            case TRIP_START:
                //增加当前点的定时器，FORCE_END_DURATION后没有点到来，强制结束行程触发窗口计算；
                ctx.registerProcessingTimeTimer(currGpsTime + TripJudgeService.FORCE_END_DURATION);

                //初始化行程状态
                tripJudge = Optional.ofNullable(tripJudge).orElse(new TripJudgeState());
                tripJudge.addCount0200(1);
                tripJudge.setLowRotationDuration(0);
                tripJudge.setLastGpsTime(currGpsTime);
                tripJudge.setStartGpsTime(currGpsTime);
                tripJudge.setLastRotation(element.getRotation());
                tripJudgeState.update(tripJudge);

                result = TriggerResult.CONTINUE;
                break;
            case TRIP_IN:
                tripJudge = Optional.ofNullable(tripJudge).orElse(new TripJudgeState());
                Long lastGpsTime = tripJudge.getLastGpsTime();

                if(null != lastGpsTime){
                    //删除上一个点的定时器，避免重复调用
                    ctx.deleteProcessingTimeTimer(lastGpsTime + TripJudgeService.FORCE_END_DURATION);
                    //增加当前点的定时器，FORCE_END_DURATION后没有点到来，强制结束行程触发窗口计算；
                    ctx.registerProcessingTimeTimer(currGpsTime + TripJudgeService.FORCE_END_DURATION);

                    //更新行程状态
                    int lastRotation = tripJudge.getLastRotation();
                    tripJudge.addCount0200(1);
                    tripJudge.setLastGpsTime(currGpsTime);
                    tripJudge.setLastRotation(element.getRotation());
                    //若当前点与上一个点的发动机转速都小于500则累计LowRotationTime，否则清零
                    tripJudge.setLowRotationDuration((lastRotation<=500 && element.getRotation()<=500) ? tripJudge.getLowRotationDuration()+(currGpsTime-lastGpsTime) : 0);
                    tripJudgeState.update(tripJudge);
                }

                result = TriggerResult.CONTINUE;
                break;
            case TRIP_END:
            case TRIP_FORCE_END:
                tripJudge = Optional.ofNullable(tripJudge).orElse(new TripJudgeState());

                //删除上一个点的定时器，避免重复调用
                Optional.ofNullable(tripJudge.getLastGpsTime()).ifPresent(lastTime ->
                    ctx.deleteProcessingTimeTimer(lastTime + TripJudgeService.FORCE_END_DURATION)
                );

                //如果行程时间 小于 最小行程时间 则清除数据
                Long startTime = Optional.ofNullable(tripJudge.getStartGpsTime()).orElse(currGpsTime);
                if(currGpsTime-startTime < 1000*tripMinDuration){
                    result = TriggerResult.PURGE;
                }
                //否则，触发窗口计算并清除数据
                else{
                    result = TriggerResult.FIRE_AND_PURGE;
                }

                //清空行程状态
                tripJudgeState.clear();
                break;
            default:
                result = TriggerResult.CONTINUE;
                break;
        }

        return result;
    }

    /**
     * 处理通过coGroup连接的两个流
     * @param element 第一个元素为0f37转换的PointProtocol, 第二个元素为0200转换的PointProtocol
     * @param ctx 上下文
     * @return 触发结果
     */
    private TriggerResult getTriggerResult(CoGroupedStreams.TaggedUnion<PointProtocol, PointProtocol> element, TriggerContext ctx) throws java.io.IOException {
        //使用0200判断行程，0F37点跳过
        if(element.isOne()){
            return TriggerResult.CONTINUE;
        }

        //数据合法性校验
        PointProtocol point = element.getTwo();
        if(!checkData(point)){
            return TriggerResult.CONTINUE;
        }

        //缓存的行程状态判断数据
        ValueState<TripJudgeState> tripJudgeState = ctx.getPartitionedState(tripJudgeDesc);
        TripJudgeState tripJudge = tripJudgeState.value();
        long currGpsTime = point.getGpsTime();

        //行程状态处理
        TriggerResult result;
        switch (tripJudgeService.tripStatus(tripJudge, point)) {
            case TRIP_START:
                //增加当前点的定时器，FORCE_END_DURATION后没有点到来，强制结束行程触发窗口计算；
                ctx.registerProcessingTimeTimer(currGpsTime + TripJudgeService.FORCE_END_DURATION);

                //初始化行程状态
                tripJudge = Optional.ofNullable(tripJudge).orElse(new TripJudgeState());
                tripJudge.addCount0200(1);
                tripJudge.setLowRotationDuration(0);
                tripJudge.setLastGpsTime(currGpsTime);
                tripJudge.setLastRotation(point.getRotation());
                tripJudgeState.update(tripJudge);

                result = TriggerResult.CONTINUE;
                break;
            case TRIP_IN:
                tripJudge = Optional.ofNullable(tripJudge).orElse(new TripJudgeState());
                Long lastGpsTime = tripJudge.getLastGpsTime();

                if(null != lastGpsTime){
                    //删除上一个点的定时器，避免重复调用
                    ctx.deleteProcessingTimeTimer(lastGpsTime + TripJudgeService.FORCE_END_DURATION);
                    //增加当前点的定时器，FORCE_END_DURATION后没有点到来，强制结束行程触发窗口计算；
                    ctx.registerProcessingTimeTimer(currGpsTime + TripJudgeService.FORCE_END_DURATION);

                    //更新行程状态
                    int lastRotation = tripJudge.getLastRotation();
                    tripJudge.addCount0200(1);
                    tripJudge.setLastGpsTime(currGpsTime);
                    tripJudge.setLastRotation(point.getRotation());
                    //若当前点与上一个点的发动机转速都小于500则累计LowRotationTime，否则清零
                    tripJudge.setLowRotationDuration((lastRotation<=500 && point.getRotation()<=500) ? tripJudge.getLowRotationDuration()+(currGpsTime-lastGpsTime) : 0);
                    tripJudgeState.update(tripJudge);
                }

                result = TriggerResult.CONTINUE;
                break;
            case TRIP_END:
            case TRIP_FORCE_END:
                //删除上一个点的定时器，避免重复调用
                Optional.ofNullable(tripJudge).ifPresent(e -> {
                    Optional.ofNullable(e.getLastGpsTime()).ifPresent(lastTime ->
                        ctx.deleteProcessingTimeTimer(lastTime + TripJudgeService.FORCE_END_DURATION)
                    );
                });

                //清空行程状态
                tripJudgeState.clear();

                //触发窗口计算,并清除数据
                result = TriggerResult.FIRE_AND_PURGE;
                break;
            default:
                result = TriggerResult.CONTINUE;
                break;
        }

        return result;
    }

    /**
     * 强制结束行程的定时器
     * @param time 定时器的启动时间，为lastGpsTime + TripJudgeService.FORCE_END_DURATION
     * @param window 当前窗口
     * @param ctx    上下文
     * @return       触发结果
     */
    @Override
    public TriggerResult onProcessingTime(long time, GlobalWindow window, TriggerContext ctx) throws Exception {
        ValueState<TripJudgeState> tripJudgeState = ctx.getPartitionedState(tripJudgeDesc);
        TripJudgeState tripJudge = tripJudgeState.value();

        //超过行程强制结束时长，触发计算,并清除数据
        if(null!=tripJudge && null!=tripJudge.getLastGpsTime()){
            if (time-tripJudge.getLastGpsTime() >= TripJudgeService.FORCE_END_DURATION) {
                tripJudgeState.clear();
                return TriggerResult.FIRE_AND_PURGE;
            }
        }

        return TriggerResult.CONTINUE;
    }

    @Override
    public TriggerResult onEventTime(long time, GlobalWindow window, TriggerContext ctx) {
        return TriggerResult.CONTINUE;
    }

    @Override
    public void clear(GlobalWindow window, TriggerContext ctx) throws Exception {
        ValueState<TripJudgeState> tripJudgeState = ctx.getPartitionedState(tripJudgeDesc);
        Optional.ofNullable(tripJudgeState.value()).flatMap(e -> Optional.ofNullable(e.getLastGpsTime())).ifPresent(lastTime ->
                ctx.deleteProcessingTimeTimer(lastTime + TripJudgeService.FORCE_END_DURATION));

        tripJudgeState.clear();
    }

    /**
     * 数据合法性校验
     * @param point 公关轨迹点
     * @return 检查结果
     */
    private boolean checkData(PointProtocol point) {
        //本次gps时间大于当前时间一个小时以上，不予处理
        if (point.getGpsTime() > System.currentTimeMillis()+3600000) {
            logger.error("===本次gps时间大于当前时间一个小时以上，不予处理! (本次:{}) 通信号：{} ===", point.getGpsTime(), point.getTid());
            return false;
        }

        //发动机转速不在合理区间内
        int rotation = point.getRotation();
        if (rotation<rotationMin || rotation>rotationMax) {
            logger.error("===发动机转速不在合理区间内，不予处理!（转速：" + rotation + "rpm）通信号：{} ===",  point.getTid());
            return false;

        }

        //速度不在合理区间内
        double speedKmh = point.getSpeedKmh();
        if (speedKmh<speedMin || speedKmh>speedMax) {
            logger.error("===速度不在合理区间内，不予处理!（速度：" + speedKmh + "km/h）通信号：{} ===",  point.getTid());
            return false;
        }

        return true;
    }

}
